package com.intelligence.device.cloud.messageConsumer;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.intelligence.common.system.AuthContextUtil;
import com.intelligence.common.system.MessageUtil;
import com.intelligence.device.cloud.config.LogAnnotation;
import com.intelligence.device.cloud.mapper.DeviceGatherMapper;
import com.intelligence.entity.pojo.*;
import com.intelligence.openfeign.service.deivce.*;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;

/**
 * 该采集模块实现了两个基地分别1、2台气象站数据的采集
 * activemq消息接收,监听开发快传递过来的传感器数据,类型为byte[]
 */
@Service
public class MessageConsumerService {

    //从redis中获取数据的阈值,阈值放在数据库中,后台可以进行维护,维护完毕后同步数据库和redis
    //在批量导入的同时对阈值进行比较,有问题直接发短信,发邮件
    //邮件短信的报警联系人也可以在后台进行设置
    @Autowired
    private RedisTemplate redisTemplate;

    //批量插入
    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private DeviceThresholdFeignService deviceThresholdFeignService;
    @Autowired
    private DeviceFeignService deviceFeignService;
    @Autowired
    private DeviceThresholdInfoFeignService deviceThresholdInfoFeignService;
    @Autowired
    private DeviceLogFeignService deviceLogFeignService;

    @Autowired
    private MonpointsDeviceFeignService monpointsDeviceFeignService;
    @Autowired
    private MeasuermentUnitFeignService measuermentUnitFeignService;

    /**
     * 监听该采集队列
     * 每两个是一组数据，其中7、21、23 是四个字节为一组数据
     *
     * @param message 接收到的消息
     */
    @Transactional(rollbackFor = Exception.class)
    @LogAnnotation
    @RabbitListener(queues = "device-cloud-queue")
    public void receiveMessage(Message message) {    // 进行消息接收处理
        byte[] payload = message.getBody();
        String s = new String(payload);
        System.err.println("【*** 接收消息转化成string ***】" + s);
        //先将流转化成字符串，然后再去判断是心跳还是采集数据
        //分割字符串----》变成一个string类型的数组
        String[] strings = MessageUtil.handleStr2ArrStr(s);
        //将字符串数组变成int数组
        Integer[] integers = MessageUtil.handleStr2ArrInt(strings);
        System.out.println("当前接收到的数据转换成数组之后的值------>");
        for (int i = 0; i < integers.length; i++) {
            System.out.print("第" + i + "个数据---->" + integers[i]);
            System.out.println("---------------------------------");
        }
        //检测数据包是否存在问题
        boolean isValid = checkDataPacketIsValid(integers);
        if (isValid) {
            //从redis中将设备的阈值取出来,此处redis可能为空
            Map<Long, DeviceThreshold> deviceThreshold = redisTemplate.opsForHash().entries("device-threshold");
            if (ObjectUtil.isEmpty(deviceThreshold)) {
                List<DeviceThreshold> list = this.deviceThresholdFeignService.listDeviceThreshold().getData();
                for (DeviceThreshold d : list) {
                    redisTemplate.opsForHash().put("device-threshold", d.getDeviceid(), d);
                }
            }

            Date date = new Date();
            //批量操作
            SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH);
            DeviceGatherMapper deviceGatherMapper = sqlSession.getMapper(DeviceGatherMapper.class);

            //空气温度传感器
            DeviceGather deviceGather = new DeviceGather();
            deviceGather.setGathertime(date);
            deviceGather.setCreatetime(date);
            deviceGather.setUpdatetime(date);

            //判断空气温度是否为负数
//        Integer basicData = 0;
//        if (strings[7].startsWith("f")) {
//            basicData = ((Integer.valueOf(strings[7] + strings[8], 16).shortValue()) / 256 - 1);
//        } else {
//            basicData = Integer.valueOf(strings[7] + strings[8], 16);
//        }

            //遍历我们从消息队列中拿出来的数据，进行数据存储
            for (int i = 0; i < integers.length / 2; i++) {
                //根据设备id查询设备信息
                Device device = deviceFeignService.getOneDevice(integers[i * 2]).getData();
                deviceGather.setDevicesn(device.getSn());
                deviceGather.setDevicename(device.getName());
                deviceGather.setDeviceid(integers[i * 2]);
                deviceGather.setDevicetype(device.getTypeid() + "");
                MonpointsDevice deviceid = monpointsDeviceFeignService.getMonpointsDevice(integers[i * 2].longValue()).getData();
                //判断阈值
                float f;
                if (i == 0) {
                    f = integers[1] / 10F;
                } else {
                    f = integers[(i * 2) + 1].floatValue();
                }
                boolean b = compareThreshold(device, f, deviceid.getMid().longValue());
                if (!b) {
                    System.err.println("------------------->此时的设备主键---->" + device.getId());
                    deviceGather.setStatus(1);
                    //报警处理步骤
                    Object deviceErrorRecords = this.redisTemplate.opsForHash().get("device-error-records", device.getId());
                    if (ObjectUtil.isEmpty(deviceErrorRecords)) {
                        //当前设备第一次报警，设置一个初始值并放入到redis中
                        this.redisTemplate.opsForHash().put("device-error-records", device.getId(), 0);
                    } else {
                        Integer deviceErrorRecordsInteger = (Integer) deviceErrorRecords;
                        if (deviceErrorRecordsInteger >= 3) {
                            //发送短信
                            System.out.println("设备检测多次不达标，请检查设备是否出现了故障");
                            this.redisTemplate.opsForHash().put("device-error-records", device.getId(), 0);
                        } else {
                            deviceErrorRecordsInteger++;
                            this.redisTemplate.opsForHash().put("device-error-records", device.getId(), deviceErrorRecordsInteger);
                        }
                    }
                } else {
                    deviceGather.setStatus(0);
                }
                deviceGather.setBasicdata((double) f);
//                deviceGather.setBasicdata(integers[(i * 2) + 1].doubleValue());
                deviceGather.setMonpointsid(deviceid.getMid());
                DeviceThreshold deviceId = this.deviceThresholdFeignService.getDeviceThreshold(integers[i * 2].longValue()).getData();
                deviceGather.setMeasurementunitid(deviceId.getMeasurementid().intValue());
                MeasurementUnit byId = this.measuermentUnitFeignService.getMeasurementUnit(deviceId.getMeasurementid()).getData();
                deviceGather.setMeasurementunitname(byId.getEnname());
                deviceGather.setMeasureunittype(byId.getUnittypes());
                deviceGatherMapper.insert(deviceGather);
            }

            sqlSession.flushStatements();
        }
    }


    /**
     * @param device    从redis中取出来的设备对象,包含阈值范围 10-100
     * @param basicData 要校验的采集数据,如果数据超出阈值范围就发消息报警
     */
    private boolean compareThreshold(Device device, Float basicData, Long mid) {
        DeviceThreshold deviceThreshold = (DeviceThreshold) redisTemplate.opsForHash().get("device-threshold", device.getId().longValue());
        //如果小于最小值  大于最大值  那么证明该数组已经超出了阈值，保存异常消息
        if (!ObjectUtil.isEmpty(deviceThreshold)) {
            if (basicData < deviceThreshold.getMinthreshold().floatValue() || basicData > deviceThreshold.getMaxthreshold().floatValue()) {
                DeviceThresholdInfo deviceThresholdInfo = new DeviceThresholdInfo();
                deviceThresholdInfo.setDeviceid(device.getId().longValue());
                deviceThresholdInfo.setGathervalue(basicData.longValue());
                deviceThresholdInfo.setMonpointid(mid);
                //交给rabbitMQ来进行异常数据的保存
                this.deviceThresholdInfoFeignService.saveDeviceThresholdInfo(deviceThresholdInfo);
//                rabbitTemplate.convertAndSend("gather-error-exchange", "gather-error-routingKey", JSON.toJSONString(deviceLog));
                return false;
            }
        } else {
            return true;
        }
        return true;
    }

    @RabbitListener(queues = "gather-error-queue")
    public void deviceThresholdErrorInfo(byte[] payload) {
        DeviceLog deviceLog = JSON.parseObject(payload.toString(), DeviceLog.class);
        System.out.println("本次接收到的错误数据采集信息是:" + deviceLog);
        this.deviceLogFeignService.saveDeviceLog(deviceLog);
    }

    /**
     * 对数据包进行校验
     *
     * @param integers
     * @return
     */
    @LogAnnotation
    private boolean checkDataPacketIsValid(Integer[] integers) {
        boolean flag = true;
        if (integers.length < 1) {
            flag = false;
            throw new RuntimeException("数据包长度不足");
        }

//        if (integers[0] != 0x5a && integers[1] != 0x5a) {
//            flag = false;
//            throw new RuntimeException("数据包头格式不正确");
//        }

        Integer dataLength = integers[3];//数据包除了校验位的长度
        if (integers.length != 24) {
            flag = false;
            throw new RuntimeException("数据包长度不正确");
        }
        //获取所有除了校验位所有字节相加取低字节的值
//        Integer dataSum = 0;
//        for (int i = 0; i < integers.length - 1; i++) {
//            dataSum += integers[i];
//        }
//        if (integers[dataLength - 1] != (dataSum & 0xFF)) {
//            flag = false;
//            throw new RuntimeException("数据包内容校验未通过");
//        }
        return flag;
    }

}
